package com.inspur.springkafka.stream1;

import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.PunctuationType;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;

public class WordCountProcessor implements Processor<String,String> {
    private ProcessorContext context;
    private HashMap<String,Integer> wordPair=null;
    @Override
    public void init(ProcessorContext context) {
        System.out.println("-----init----");
        this.context=context;
        wordPair=new HashMap<>();
        //定时输出结果
        context.schedule(Duration.ofSeconds(1), PunctuationType.WALL_CLOCK_TIME,(ts)->{
            for (Map.Entry<String, Integer> entry : wordPair.entrySet()) {
                context.forward(entry.getKey(),entry.getValue());
            }
        });
    }
    @Override
    public void process(String key, String value) {
        String[] words = value.split("\\W+");
        for (int i = 0; i < words.length; i++) {
            int count=0;
            if(wordPair.containsKey(words[i])){
                count=wordPair.get(words[i]);
            }
            count+=1;
            wordPair.put(words[i],count);
        }
    }
    @Override
    public void close() {

    }
}
